package org.rocky.consumer.service;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.PropertySource;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Arrays;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author Rocky
 * @version 1.0
 * @date 2020/3/19 15:16
 */
@Component
@PropertySource("classpath:queues.properties")
public class ConsumeDelay {

    @Autowired
    Map<String,Integer> keys;

    @RabbitHandler
    @RabbitListener(queues = "DELAY_QUEUE", containerFactory="rabbitListenerContainerFactory"  )
    public void processDelay(Message message, Channel channel) throws IOException {


        System.out.printf("Delay Queue received msg : %s%n", new String(message.getBody()));
        String msg = new String(message.getBody());
       if(!keys.containsKey(msg)){

            channel.basicPublish("DELAY_EXCHANGE","rocky.exchange",new AMQP.BasicProperties(),message.getBody());
            keys.put(msg,1);
        }else {
            if(keys.get(msg)>5){

                channel.basicPublish("DEADLINE_EXCHANGE","#",new AMQP.BasicProperties(),message.getBody());
                keys.remove(msg);
            }else {
                if(keys.get(msg)==3){
                    //ACK
                    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                }

                keys.put(msg,keys.get(msg)+1);
                channel.basicPublish("DELAY_EXCHANGE","rocky.exchange",new AMQP.BasicProperties(),message.getBody());
            }
        }
        //拒绝 进入死信队列
        //channel.basicReject(message.getMessageProperties().getDeliveryTag(),false);
        //重回 慎用
        //channel.basicPublish("TTL_EXCHANGE","rocky.ttl",new AMQP.BasicProperties(),message.getBody());

    }
}
